package com.eastfair.messagecenter.duty;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.eastfair.auth.api.UserAccountServiceFeign;
import com.eastfair.auth.dto.UserAccountDto;
import com.eastfair.auth.enumeration.OrganizationOrgTypeEnum;
import com.eastfair.auth.vo.UserAccountVo;
import com.eastfair.constant.BusinessConstant;
import com.eastfair.core.base.R;
import com.eastfair.core.context.ContextUtil;
import com.eastfair.core.exception.MessageBizException;
import com.eastfair.messagecenter.ContentModel;
import com.eastfair.messagecenter.domain.TaskInfo;
import com.eastfair.messagecenter.dto.MsgDTO;
import com.eastfair.messagecenter.entity.InnerTemplate;
import com.eastfair.messagecenter.entity.Msg;
import com.eastfair.messagecenter.entity.SmsTemplate;
import com.eastfair.messagecenter.enumeration.BusinessCode;
import com.eastfair.messagecenter.enumeration.ChannelType;
import com.eastfair.messagecenter.enumeration.SendStatus;
import com.eastfair.messagecenter.model.InnerContentModel;
import com.eastfair.messagecenter.model.SmsContentModel;
import com.eastfair.messagecenter.pipeline.BusinessProcess;
import com.eastfair.messagecenter.pipeline.ProcessContext;
import com.eastfair.messagecenter.pipeline.SendTaskModel;
import com.eastfair.messagecenter.point.ReadySendPoint;
import com.eastfair.messagecenter.service.InnerTemplateService;
import com.eastfair.messagecenter.service.MsgService;
import com.eastfair.messagecenter.service.SmsTemplateService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 组装任务task数据
 * 功能：
 * 1组装消息信息
 * 2消息入库
 * 3遇到要发送短信和邮箱时，则组装taskinfo
 *
 * @author ligang
 * @date 2021/11/22
 * @description 组装task任务
 */
@Slf4j
@Service
public class AssembleTaskAction implements BusinessProcess<SendTaskModel> {

    @Autowired
    private UserAccountServiceFeign userAccountServiceFeign;


    @Autowired
    private SmsTemplateService smsTemplateService;

    @Autowired
    private InnerTemplateService innerTemplateService;


    @Autowired
    private MsgService msgService;

    /**
     * 最大200人数一个任务包
     */
    private static final Integer BATCH_RECEIVER_SIZE = 200;

    @Override
    public void process(ProcessContext<SendTaskModel> context) {
        SendTaskModel sendTaskModel = context.getProcessModel();
        ReadySendPoint readySendPoint = sendTaskModel.getReadySendPoint();
        List<TaskInfo> taskInfoList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(readySendPoint.getSendMobileList())) {
            //有手机号，则只发送手机号
            taskInfoList.addAll(assembleTaskInfo(readySendPoint, null));
        } else {
            //用户组
            List<List<Long>> userIdsOfList = assembleUserId(readySendPoint);
            for (int i = 0; i < userIdsOfList.size(); i++) {
                taskInfoList.addAll(assembleTaskInfo(readySendPoint, userIdsOfList.get(i)));
            }
        }
        //组装任务
        sendTaskModel.setTaskInfoList(taskInfoList);
        context.setNeedBreak(false);
        context.setProcessModel(sendTaskModel);
    }

    /**
     * 组装 TaskInfo 任务消息
     *
     * @param readySendPoint
     * @param userIds
     */
    private List<TaskInfo> assembleTaskInfo(ReadySendPoint readySendPoint, List<Long> userIds) {
        List<TaskInfo> taskInfoList = new ArrayList<>();
        //组装 发送内容
        List<ContentModel> contentModelList = assembleContentModel(readySendPoint, readySendPoint.getContentModelList());
        //批量添加
        List<Msg> msgList = null;
        if (CollectionUtil.isNotEmpty(readySendPoint.getSendMobileList())) {
            msgList = batchAddmsgOfMobile(readySendPoint.getSendMobileList(), readySendPoint, contentModelList);
        } else {
            msgList = batchAddmsg(userIds, readySendPoint, contentModelList);
        }
        //过滤掉实时发送
        msgList = msgList.stream().filter(s -> BusinessConstant.YES == s.getIsSendNow()).collect(Collectors.toList());
        //站内信发送
        innerSend(msgList, readySendPoint);
        //组装短信和邮件的发送任务
        taskInfoList = assmbileTask(msgList, readySendPoint, userIds, contentModelList);
        return taskInfoList;

    }

    public void innerSend(List<Msg> msgList, ReadySendPoint readySendPoint) {
        if (ChannelType.INNER_IM.equals(readySendPoint.getChannelType())) {
            for (Msg msg : msgList) {
                if (msg.getIsInnerMsgSend() != null && BusinessConstant.YES == msg.getIsInnerMsgSend() && BusinessConstant.YES == msg.getIsSendNow()) {
                    msg.setIsInnerMsgSendSucc(BusinessConstant.YES);
                    msg.setSendTime(LocalDateTime.now());
//                msgService.updateById(msg);
                }
            }
            msgService.updateBatchById(msgList);
        }
    }

    public List<TaskInfo> assmbileTask(List<Msg> msgList, ReadySendPoint readySendPoint, List<Long> userIds, List<ContentModel> contentModelList) {
        List<TaskInfo> taskInfoList = new ArrayList<>();
        if (CollectionUtil.isNotEmpty(msgList)) {
            //目前只有邮箱和短信发送需要添加发送任务
            // TODO: 2022/9/9 1 填充message表数据，2 组装发送内容 3记录消息id，用于记录回传信息
            for (ContentModel contentModel : contentModelList) {
                if (ChannelType.SMS.equals(readySendPoint.getChannelType())) {
//                  mobiles = assmbileMobiles(readySendPoint, userIds);
                    //过滤掉校验失败的短信
                    List<Msg> sendMsgList = msgList.stream()
                            .filter(s -> s.getIsNoteSend() == BusinessConstant.YES && !SendStatus.VERIFY_FAIL.eq(s.getSendStatus()))
                            .collect(Collectors.toList());
                    if (CollectionUtil.isNotEmpty(sendMsgList)) {
                        SmsContentModel smsContentModel = (SmsContentModel) contentModel;
                        TaskInfo taskInfo = new TaskInfo();
                        taskInfo.setContentModel(smsContentModel);
                        taskInfo.setMsgIds(sendMsgList.stream().map(Msg::getId).collect(Collectors.toList()));
                        taskInfo.setReceiverUserIds(userIds);
                        taskInfo.setSendChannel(readySendPoint.getChannelType());
                        taskInfo.setMsgType(readySendPoint.getMsgType());
                        taskInfo.setMobiles(sendMsgList.stream()
                                .map(Msg::getMobile).collect(Collectors.toList()));
                        taskInfoList.add(taskInfo);
                    }

                } else if (ChannelType.INNER_IM.equals(readySendPoint.getChannelType())) {
                }
            }
        }
        return taskInfoList;
    }

    private List<String> assmbileMobiles(ReadySendPoint readySendPoint, List<Long> userIds) {
        if (CollectionUtil.isNotEmpty(userIds)) {
            //目前只有邮箱和短信发送需要添加发送任务
            UserAccountDto userAccountDto = new UserAccountDto();
            userAccountDto.setUserIds(userIds);
            R<List<UserAccountVo>> listR = userAccountServiceFeign.querySimpleByIds(userAccountDto);
            if (listR.getIsSuccess() && CollectionUtil.isNotEmpty(listR.getData())) {
                return listR.getData().stream().map(UserAccountVo::getMobile).collect(Collectors.toList());
            }
        } else if (CollectionUtil.isNotEmpty(readySendPoint.getSendMobileList())) {
            List<String> list2 = new ArrayList<String>();
            list2.addAll(readySendPoint.getSendMobileList());
            return list2;
        }
        return null;
    }

    private List<Msg> batchAddmsgOfMobile(Set<String> mobiles, ReadySendPoint readySendPoint, List<ContentModel> contentModelList) {
        List<Msg> msgList = new ArrayList<>();
        Set<String> noSendMoibleSet = null;
        if (mobiles != null) {
            noSendMoibleSet = filterNoSendOfMobile(new ArrayList<String>(mobiles));
        }
        for (String mobile : mobiles) {
            Msg msg = new Msg();
            msg.setBusinessId(readySendPoint.getBusinessId());
            msg.setBusinessType(readySendPoint.getBusinessType());
            msg.setMsgType(readySendPoint.getMsgType());
            msg.setMsgContentType(readySendPoint.getMsgContentType());
            if (readySendPoint.getIsSendAll() != null && readySendPoint.getIsSendAll()) {
                msg.setIsSendUserAll(BusinessConstant.YES);
            }
            msg.setIsSendNow(readySendPoint.getIsSendNow());
            msg.setPredictSendTime(readySendPoint.getPredictSendTime());
            msg.setMobile(mobile);
            for (ContentModel contentModel : contentModelList) {
                if (ChannelType.SMS.equals(readySendPoint.getChannelType())) {
                    SmsContentModel smsContentModel = (SmsContentModel) contentModel;
                    msg.setContent(smsContentModel.getContent());
                    msg.setIsNoteSend(BusinessConstant.YES);
                    msg.setHandlerUrl(smsContentModel.getUrl());
                    msg.setHandlerParams(JSONObject.toJSONString(smsContentModel.getContentMap()));
                    msg.setNoteTemplateId(smsContentModel.getTemplateId());
                    msg.setSendStatus(SendStatus.WAIT);
                    msg.setIsRead(BusinessConstant.YES);
                    //发送限制
                    if (msg.getMobile() != null && noSendMoibleSet != null && noSendMoibleSet.contains(msg.getMobile())) {
                        msg.setSendStatus(SendStatus.VERIFY_FAIL);
                    }
                }
            }
            msgList.add(msg);
        }
        //批量添加
        msgService.addBatch(msgList);
        return msgList;
    }


    private List<Msg> batchAddmsg(List<Long> userIds, ReadySendPoint readySendPoint, List<ContentModel> contentModelList) {
        List<Msg> msgList = new ArrayList<>();
        UserAccountDto userAccountDto = new UserAccountDto();
        userAccountDto.setUserIds(userIds);
        R<Map<Long, String>> mapR = userAccountServiceFeign.queryMobiles(userAccountDto);
        Map<Long, String> moibleMap = mapR.getData();
        Set<String> noSendMoibleSet = null;
        if (moibleMap != null) {
            List<String> mobileList = new ArrayList<>();
            for (String mobile : moibleMap.values()) {
                mobileList.add(mobile);
            }
            noSendMoibleSet = filterNoSendOfMobile(mobileList);
        }
        for (Long userId : userIds) {
            Msg msg = new Msg();
            msg.setBusinessId(readySendPoint.getBusinessId());
            msg.setBusinessType(readySendPoint.getBusinessType());
            msg.setMsgType(readySendPoint.getMsgType());
            msg.setMsgContentType(readySendPoint.getMsgContentType());
            if (readySendPoint.getIsSendAll() != null && readySendPoint.getIsSendAll()) {
                msg.setIsSendUserAll(BusinessConstant.YES);
            }
            msg.setIsSendNow(readySendPoint.getIsSendNow());
            msg.setPredictSendTime(readySendPoint.getPredictSendTime());
            msg.setUserId(userId);
            for (ContentModel contentModel : contentModelList) {
                if (ChannelType.SMS.equals(readySendPoint.getChannelType())) {
                    SmsContentModel smsContentModel = (SmsContentModel) contentModel;
                    msg.setContent(smsContentModel.getContent());
                    msg.setIsNoteSend(BusinessConstant.YES);
                    msg.setHandlerUrl(smsContentModel.getUrl());
                    msg.setHandlerParams(JSONObject.toJSONString(smsContentModel.getContentMap()));
                    msg.setNoteTemplateId(smsContentModel.getTemplateId());
                    msg.setSendStatus(SendStatus.WAIT);
                    msg.setIsRead(BusinessConstant.YES);
                    if (moibleMap != null && StringUtils.isNotBlank(moibleMap.get(userId))) {
                        msg.setMobile(moibleMap.get(userId));
                    }
                    //发送限制
                    if (msg.getMobile() != null && noSendMoibleSet != null && noSendMoibleSet.contains(msg.getMobile())) {
                        msg.setSendStatus(SendStatus.VERIFY_FAIL);
                    }
                } else if (ChannelType.INNER_IM.equals(readySendPoint.getChannelType())) {
                    InnerContentModel innerContentModel = (InnerContentModel) contentModel;
                    msg.setContent(innerContentModel.getContent());
                    msg.setHandlerUrl(innerContentModel.getUrl());
                    msg.setHandlerParams(JSONObject.toJSONString(innerContentModel.getContentMap()));
                    msg.setInnerMsgTemplateId(innerContentModel.getTemplateId());
                    msg.setSendStatus(SendStatus.SEND_SUCC);
                    msg.setIsInnerMsgSendSucc(BusinessConstant.YES);
                    msg.setIsRead(BusinessConstant.NO);
                    msg.setIsInnerMsgSend(BusinessConstant.YES);
                    if (readySendPoint.getPredictSendTime() != null) {
                        msg.setSendStatus(SendStatus.WAIT);
                    }
                }
            }
            msgList.add(msg);
        }
        //批量添加
        msgService.addBatch(msgList);
        return msgList;
    }

    /**
     * 过滤掉不能发送的手机号，
     * 和 10秒只能有一条
     *
     * @param mobiles
     * @return
     */
    public Set<String> filterNoSendOfMobile(List<String> mobiles) {
        Set<String> set = new HashSet<>();
        MsgDTO msgDTO = new MsgDTO();
        msgDTO.setMobiles(mobiles);
        msgDTO.setSendTimeGe(LocalDateTime.now().minusMinutes(1));
        msgDTO.setIsNoteSend(BusinessConstant.YES);
        msgDTO.setSendStatusList(Arrays.asList(SendStatus.SEND_SUCC, SendStatus.SENDING, SendStatus.WAIT));
        List<Msg> msgList = msgService.list(msgDTO, null);
        if (CollectionUtil.isNotEmpty(msgList)) {
            for (Msg msg : msgList) {
                if (StringUtils.isNotBlank(msg.getMobile())) {
                    set.add(msg.getMobile());
                }
            }
        }
        return set;
    }


    /**
     * 组装 发送内容
     *
     * @param readySendPoint
     * @param contentModelList
     */
    private List<ContentModel> assembleContentModel(ReadySendPoint readySendPoint, List<ContentModel> contentModelList) {
        List<ContentModel> result = new ArrayList<>();
        for (ContentModel contentModel : contentModelList) {
            if (ChannelType.SMS.equals(readySendPoint.getChannelType())) {
                SmsContentModel smsContentModel = JSONObject.parseObject(JSONObject.toJSONString(contentModel), SmsContentModel.class);
                SmsTemplate smsTemplate = smsTemplateService.getByIdCache(smsContentModel.getTemplateId());
                String content = smsTemplate.getContent();
                //初始化参数
                Map<String, Object> map = JSONObject.parseObject(smsTemplate.getTemplateParams(), Map.class);
                content = convertMap(content, map);
                //组装入参的数据
                content = convertMap(content, smsContentModel.getContentMap());
                smsContentModel.setContent(content);
                if (map != null) {
                    smsContentModel.setUrl(map.get("url") != null ? map.get("url").toString() : null);
                }
                result.add(smsContentModel);
            } else if (ChannelType.INNER_IM.equals(readySendPoint.getChannelType())) {
                InnerContentModel innerContentModel = JSONObject.parseObject(JSONObject.toJSONString(contentModel), InnerContentModel.class);
                InnerTemplate innerTemplate = innerTemplateService.getByIdCache(innerContentModel.getTemplateId());
                String content = innerTemplate.getContent();
                //初始化参数
                Map<String, Object> map = JSONObject.parseObject(innerTemplate.getTemplateParams(), Map.class);
                content = convertMap(content, map);
                //组装入参的数据
                content = convertMap(content, innerContentModel.getContentMap());
                innerContentModel.setContent(content);
                if (map != null) {
                    innerContentModel.setUrl(map.get("url") != null ? map.get("url").toString() : null);
                }
                result.add(innerContentModel);
            }
        }
        return result;
    }


    public String convertMap(String content, Map<String, Object> map) {
        if (map != null) {
            for (String key : map.keySet()) {
                content = content.replaceAll("\\[\\[" + key + "\\]\\]", map.get(key).toString());
            }
        }
        return content;
    }

    /**
     * 组装用户id，通过组织id和全部
     * 将用户id集合拆分为200个人的一个组，避免全部一起发带来的性能影响
     *
     * @return
     */
    public List<List<Long>> assembleUserId(ReadySendPoint readySendPoint) {
        List<Long> userIds = new ArrayList<>();
        //查全部
        if (readySendPoint.getIsSendAll() != null && readySendPoint.getIsSendAll()) {
            UserAccountDto userAccountDto = new UserAccountDto();
            userAccountDto.setIsFindAll(true);
            R<List<Long>> listR = userAccountServiceFeign.queryIds(userAccountDto);
            log.info("userid is {} ", JSONObject.toJSONString(listR));
            if (listR.getIsSuccess()) {
                userIds = listR.getData();
            }
        } else if (CollectionUtil.isNotEmpty(readySendPoint.getOrgIdList())) {
            UserAccountDto userAccountDto = new UserAccountDto();
            List<Long> orgIds = new ArrayList<>();
            orgIds.addAll(readySendPoint.getOrgIdList());
            userAccountDto.setOrgIds(orgIds);
            R<List<Long>> listR = userAccountServiceFeign.queryIds(userAccountDto);
            if (listR.getIsSuccess()) {
                userIds = listR.getData();
            }
        } else if (CollectionUtil.isNotEmpty(readySendPoint.getSenUserIdList())) {
            userIds.addAll(readySendPoint.getSenUserIdList());
        }

        if (CollectionUtil.isEmpty(userIds)) {
            throw MessageBizException.wrap("发送人为空{}，", JSONObject.toJSONString(readySendPoint));
        }

        //200个人一个任务组
        int limit = userIds.size() / BATCH_RECEIVER_SIZE + 1;
        List<Long> finalUserIds = userIds;
        return Stream.iterate(0, n -> n + 1)
                .limit(limit)
                .parallel()
                .map(a -> finalUserIds.parallelStream().skip(a * BATCH_RECEIVER_SIZE).limit(BATCH_RECEIVER_SIZE).collect(Collectors.toList()))
                .filter(b -> !b.isEmpty())
                .collect(Collectors.toList());

    }
}
